﻿/*
Copyright (C) 2009  Tomasz Chrzanowski

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>
*/

using System;
using System.ComponentModel;
using System.Data;
using System.Data.SqlClient;
using System.Diagnostics;
using System.IO;
using System.Text;
using System.Transactions;

namespace Com.TomCdc.ChangeDataCapture
{
    public class CdcSchemaInstaller
    {
        /// <summary>
        /// The place holder of a database name in a template file.
        /// </summary>
        private const string PLACE_HOLDER_DATABASE_NAME = "{dbName}";

        /// <summary>
        /// The place holder of a table identifier in a template file.
        /// </summary>
        private const string PLACE_HOLDER_TABLE_IDENTIFIER = "{tableIdentifier}";
        
        /// <summary>
        /// The place holder of a table schema in a template file.
        /// </summary>
        private const string PLACE_HOLDER_TABLE_SCHEMA = "{tableSchema}";

        /// <summary>
        /// The place holder of a table name in a template file.
        /// </summary>
        private const string PLACE_HOLDER_TABLE_NAME = "{tableName}";
    
        /// <summary>
        /// The place holder of the columns list in a template file.
        /// </summary>
        private const string PLACE_HOLDER_CASE_RELEVANT_COLUMNS_LIST = "{columnsList}";

        /// <summary>
        /// The place holder of the trigger prefix in a template file.
        /// </summary>
        private const string PLACE_HOLDER_TRIGGER_PREFIX = "{triggerPrefix}";

        /// <summary>
        /// The place holder of the trigger columns bitmask in a template file.
        /// </summary>
        private const string PLACE_HOLDER_COLUMNS_BITMASK = "{columnsBitMask}";

        /// <summary>
        /// The place holder of the primary key column of tracked table.
        /// </summary>
        private const string PLACE_HOLDER_SOURCE_ROW_ID = "{sourceRowId}";

        /// <summary>
        /// The trigger prefix.
        /// </summary>
        private const string TRIGGER_PREFIX = "trCDC__";

        /// <summary>
        /// The intent used in formating of a sql script.
        /// </summary>
        private const string INTENT = "    ";

        /// <summary>
        /// The prefix of a column informing whether the column value has changed or not.
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX = "isChanged__";

        /// <summary>
        /// The prefix of a column new value.
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX_INS = "ins__";

        /// <summary>
        /// The prefix of a column old value.
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX_DEL = "del__";

        /// <summary>
        /// The sql connection string bilder initialized with a connection string to monitored database.
        /// </summary>
        private readonly SqlConnectionStringBuilder _monitorDbConnStr;

        /// <summary>
        /// The sql connection string bilder initialized with a connection string to stage database.
        /// </summary>
        private readonly SqlConnectionStringBuilder _stageDbConnStr;

        /// <summary>
        /// The sql connection string bilder initialized with a connection string to store database.
        /// </summary>
        private readonly SqlConnectionStringBuilder _storeDbConnStr;

        /// <summary>
        /// The BackgroundWorker object used to perform asyncronous execution of <see cref="CdcSchemaInstaller"/>
        /// class methods by consuming object.
        /// </summary>
        private readonly BackgroundWorker _backWorker;

        /// <summary>
        /// Initializes a new instance of <see cref="DatabaseSchemaInstaller"/> class.
        /// </summary>
        /// <param name="monitorDbConnStr">The sql connection string to monitored database.</param>
        /// <param name="stageDbConnStr">The sql connection string to stage database.</param>
        /// <param name="storeDbConnStr">The sql connection string to store database.</param>
        public CdcSchemaInstaller(string monitorDbConnStr, string stageDbConnStr, string storeDbConnStr)
        {
            // initializes sql connection string builders with connection strings of affected by schema
            // change databases
            _monitorDbConnStr = new SqlConnectionStringBuilder(monitorDbConnStr);
            _stageDbConnStr = new SqlConnectionStringBuilder(stageDbConnStr);
            _storeDbConnStr = new SqlConnectionStringBuilder(storeDbConnStr);
        }

        /// <summary>
        /// Initializes a new instance of <see cref="DatabaseSchemaInstaller"/> class.
        /// </summary>
        /// <param name="monitorDbConnStr">The sql connection string to monitored database.</param>
        /// <param name="stageDbConnStr">The sql connection string to stage database.</param>
        /// <param name="storeDbConnStr">The sql connection string to store database.</param>
        /// <param name="backWorker">The background worked used by consumer object to execute async calls.</param>
        public CdcSchemaInstaller(string monitorDbConnStr, string stageDbConnStr, string storeDbConnStr,
            BackgroundWorker backWorker)
        {
            _backWorker = backWorker;
            
            // initializes sql connection string builders with connection strings of affected by schema
            // change databases
            _monitorDbConnStr = new SqlConnectionStringBuilder(monitorDbConnStr);
            _stageDbConnStr = new SqlConnectionStringBuilder(stageDbConnStr);
            _storeDbConnStr = new SqlConnectionStringBuilder(storeDbConnStr);
        }

        /// <summary>
        /// Builds sql statement dropping changes tracking trigger from the source table.
        /// </summary>
        /// <param name="tableNames">The table tableIdentifier to drop the trigger from.</param>
        /// <param name="template">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildDropTriggerSql(string tableSchema, string tableIdentifier, string template)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableSchema))
            {
                throw new ArgumentException("The tableSchema argument is null or empty.");
            }

            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableIdentifier argument is null or empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            template = template.Replace(PLACE_HOLDER_TABLE_SCHEMA, tableSchema);
            template = template.Replace(PLACE_HOLDER_TRIGGER_PREFIX, TRIGGER_PREFIX);
            return template.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);
        }

        /// <summary>
        /// Builds sql statement dropping changes tracking trigger from the source table.
        /// </summary>
        /// <param name="tableNames">The table tableIdentifier to drop the trigger from.</param>
        /// <param name="templateFileStream">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildDropTriggerSql(string tableSchema, string tableIdentifier, StreamReader templateFileStream)
        {
            return BuildDropTriggerSql(tableSchema, tableIdentifier, templateFileStream.ReadToEnd());
        }

        /// <summary>
        /// Builds sql statement creating a trigger to capture data changes on tracked source table.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of tracked source table.</param>
        /// <param name="template">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildCaptureTriggerSql(string tableIdentifier, string tableSchema, 
            string tableName, string[] columnNames, string columnsBitMask, string template)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (columnNames == null || columnNames.Length == 0)
            {
                throw new ArgumentException("The columnNames argument is null or empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            // estimates number of characters of all columns to be inserted into tamplate
            int avgAllColsLen = columnNames.Length * 20;

            // initializes template form which result will be build
            StringBuilder result = new StringBuilder(template, template.Length + avgAllColsLen * 4);

            // initializes columns collection
            StringBuilder columnList = new StringBuilder(CommonHelper.TRIGGER_COL_LIST_TOKEN, avgAllColsLen);

            // builds columns collection
            foreach (string columnName in columnNames)
            {
                if (string.IsNullOrEmpty(columnName))
                {
                    throw new ArgumentException("One of the elements of columnNames argument is null or empty.");
                }

                columnList.Append(",[" + columnName + "]");
            }

            columnList.Append(CommonHelper.TRIGGER_COL_LIST_TOKEN);

            // replaces place holders with prepared relevant data
            result.Replace(PLACE_HOLDER_DATABASE_NAME, _stageDbConnStr.InitialCatalog);
            result.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);
            result.Replace(PLACE_HOLDER_TABLE_NAME, tableName);
            result.Replace(PLACE_HOLDER_TABLE_SCHEMA, tableSchema);
            result.Replace(PLACE_HOLDER_CASE_RELEVANT_COLUMNS_LIST, columnList.ToString());
            result.Replace(PLACE_HOLDER_TRIGGER_PREFIX, TRIGGER_PREFIX);
            result.Replace(PLACE_HOLDER_COLUMNS_BITMASK, columnsBitMask);

            return result.ToString();
        }

        /// <summary>
        /// Builds sql statement creating a trigger to capture data changes on tracked source table.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of trecked source table.</param>
        /// <param name="templateFileStream">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildCaptureTriggerSql(string tableIdentifier,string tableSchema,
            string tableName, string[] columnNames,
            string columnsBitMask, StreamReader templateFileStream)
        {
            return BuildCaptureTriggerSql(tableIdentifier, tableSchema, tableName, columnNames, columnsBitMask, templateFileStream.ReadToEnd());
        }

        public string BuildChangedDataTableSql(string tableIdentifier, string template)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            // initializes template form which result will be build
            StringBuilder result = new StringBuilder(template, template.Length + tableIdentifier.Length * 10);

            // replaces place holders with prepared relevant data
            result.Replace(PLACE_HOLDER_DATABASE_NAME, _storeDbConnStr.InitialCatalog);
            result.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);

            return result.ToString();
        }

        public string BuildChangedDataTableSql(string tableName, StreamReader templateFileStream)
        {
            return BuildChangedDataTableSql(tableName, templateFileStream.ReadToEnd());
        }


        public string BuildInsertChangedDataSpSql(string tableIdentifier, string template)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            // initializes template form which result will be build
            StringBuilder result = new StringBuilder(template, template.Length + tableIdentifier.Length * 10);

            // replaces place holders with prepared relevant data
            result.Replace(PLACE_HOLDER_DATABASE_NAME, _storeDbConnStr.InitialCatalog);
            result.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);

            return result.ToString();
        }

        public string BuildInsertChangedDataSpSql(string tableName, StreamReader templateFileStream)
        {
            return BuildInsertChangedDataSpSql(tableName, templateFileStream.ReadToEnd());
        }

        /// <summary>
        /// Builds sql statement creating two tables (inserted and deleted) in stage database for 
        /// changed records in tracked source table.
        /// The type of built table (inserted or deleted) depends on the template argument content.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of trecked source table.</param>
        /// <param name="columnTypes">The types of columns of tracked source table.</param>
        /// <param name="template">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildInsertedOrDeletedStageTableSql(string tableIdentifier, string[] columnNames, string[] columnTypes,
            string template)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (columnNames.Length != columnTypes.Length || columnNames.Length == 0)
            {
                throw new ArgumentException(
                    "The columnNames and columnTypes arguments have different number of elements or are empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            // average lenght of columns list (20 - name, 10 - type)
            int avgAllColsLen = columnNames.Length * (20 + 10);

            // initializes template form which result will be build
            StringBuilder result = new StringBuilder(template, template.Length + avgAllColsLen);

            // initializes columns collection
            StringBuilder columnList = new StringBuilder(avgAllColsLen);

            // builds column collection
            for (int colIdx = 0; colIdx < columnNames.Length; colIdx++)
            {
                if (string.IsNullOrEmpty(columnNames[colIdx]) || string.IsNullOrEmpty(columnTypes[colIdx]))
                {
                    throw new ArgumentException(
                        "One of the elements of columnNames or columnTypes arguments is null or empty.");
                }

                columnList.AppendLine(INTENT + ",[" + columnNames[colIdx] + "] " + columnTypes[colIdx]);
            }

            // replaces place holders with prepared relevant data
            result.Replace(PLACE_HOLDER_DATABASE_NAME, _stageDbConnStr.InitialCatalog);
            result.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);
            result.Replace(PLACE_HOLDER_CASE_RELEVANT_COLUMNS_LIST, columnList.ToString());

            return result.ToString();
        }

        /// <summary>
        /// Builds sql statement creating two tables (inserted and deleted) in stage database for 
        /// changed records in tracked source table.
        /// The type of built table (inserted or deleted) depends on the template argument content.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of trecked source table.</param>
        /// <param name="columnTypes">The types of columns of tracked source table.</param>
        /// <param name="templateFileStream">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildInsertedOrDeletedStageTableSql(string tableName, string[] columnNames, string[] columnTypes,
            StreamReader templateFileStream)
        {
            return BuildInsertedOrDeletedStageTableSql(tableName, columnNames, columnTypes, templateFileStream.ReadToEnd());
        }

        /// <summary>
        /// Builds sql statement creating a stored procedure which produces data required for changed data
        /// identification.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of thecked source table.</param>
        /// <param name="template">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildIdentifyChangesSpSql(string tableIdentifier, string[] columnNames, string template, string sourceRowId)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (columnNames == null || columnNames.Length == 0)
            {
                throw new ArgumentException("The columnNames argument is null or empty.");
            }

            if (string.IsNullOrEmpty(template))
            {
                throw new ArgumentException("The template argument is null or empty.");
            }

            // average lenght of columns list (20 - name)
            int avgAllColsLen = columnNames.Length * 20;

            // initializes template form which result will be build
            StringBuilder result = new StringBuilder(template, template.Length + avgAllColsLen * 8);

            // initializes columns collections
            StringBuilder columnListDiff = new StringBuilder(avgAllColsLen * 4);
            StringBuilder columnListDel = new StringBuilder(avgAllColsLen * 2);
            StringBuilder columnListIns = new StringBuilder(avgAllColsLen * 2);

            // builds columns collections
            foreach (string columnName in columnNames)
            {
                if (string.IsNullOrEmpty(columnName))
                {
                    throw new ArgumentException("One of the elements of columnNames argument is null or empty.");
                }

                // example line:  ,dbo.isDifferent(I.[Cost] ,D.[Cost]) [isChanged__Cost]
                columnListDiff.AppendLine(
                    INTENT +
                    ",dbo.isDifferent(I.[" + columnName + "] ,D.[" + columnName + "]) [" +
                    STATE_CHANGE_COL_PREFIX + columnName + "]");

                // example lise:  ,I.[Cost] [ins__Cost]
                columnListDel.AppendLine(
                    INTENT + ",D.[" + columnName + "] [" + STATE_CHANGE_COL_PREFIX_DEL + columnName + "]");

                // example line: ,D.[Cost] [del__Cost]
                columnListIns.AppendLine(
                    INTENT + ",I.[" + columnName + "] [" + STATE_CHANGE_COL_PREFIX_INS + columnName + "]");
            }

            columnListDiff.AppendLine();
            columnListDel.AppendLine();
            columnListIns.AppendLine();

            // replaces place holders with prepared relevant data
            result.Replace(PLACE_HOLDER_DATABASE_NAME, _stageDbConnStr.InitialCatalog);
            result.Replace(PLACE_HOLDER_TABLE_IDENTIFIER, tableIdentifier);
            result.Replace(PLACE_HOLDER_CASE_RELEVANT_COLUMNS_LIST,
                columnListDiff.ToString() +
                columnListDel.ToString() +
                columnListIns.ToString());
            result.Replace(PLACE_HOLDER_SOURCE_ROW_ID, sourceRowId);

            return result.ToString();
        }

        /// <summary>
        /// Builds sql statement creating a stored procedure which produces data required for changed data
        /// identification.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of thecked source table.</param>
        /// <param name="templateFileStream">The sql template used to build the DDL query statement.</param>
        /// <returns></returns>
        public string BuildIdentifyChangesSpSql(string tableName, string[] columnNames,
            StreamReader templateFileStream, string sourceRowId)
        {
            return BuildIdentifyChangesSpSql(tableName, columnNames, templateFileStream.ReadToEnd(), sourceRowId);
        }

        /// <summary>
        /// Adds table and its columns to the store database reference tables.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of thecked source table.</param>
        /// <param name="sqlCmd">The sql command under which operations will be 
        /// performed (transactioned command recomended).</param>
        public void AddTableAndColumnsToDictionary(string tableIdentifier, string[] columnNames, SqlCommand sqlCmd)
        {
            // validates input arguments
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (columnNames == null || columnNames.Length == 0)
            {
                throw new ArgumentException("The columnNames argument is null or empty.");
            }

            if (sqlCmd == null)
            {
                throw new ArgumentNullException("sqlCmd");
            }

            // validates input data in advance to avoid opening transaction and rolling it back
            // when input incorrect
            foreach (string columnName in columnNames)
            {
                if (string.IsNullOrEmpty(columnName))
                {
                    throw new ArgumentException("One of the elements of columnNames argument is null or empty.");
                }
            }

            // prepares command for adding new table
            sqlCmd.CommandText = "InsertSourceTable";
            sqlCmd.CommandType = CommandType.StoredProcedure;
            sqlCmd.Parameters.Clear();
            sqlCmd.Parameters.Add(new SqlParameter("@TableName", tableIdentifier));
            sqlCmd.Parameters.Add(new SqlParameter("@TableId", SqlDbType.Int));
            sqlCmd.Parameters["@TableId"].Direction = ParameterDirection.Output;

            int affected = sqlCmd.ExecuteNonQuery();
            Debug.WriteLine("Added new table " + tableIdentifier + " to store:" + affected);

            // gets the id of inserted row
            int tableId = (int)sqlCmd.Parameters["@TableId"].Value;

            // prepares command for adding new column
            sqlCmd.CommandText = "InsertSourceColumn";
            sqlCmd.Parameters.Clear();
            sqlCmd.Parameters.Add(new SqlParameter("@TableId", tableId));
            sqlCmd.Parameters.Add(new SqlParameter("@ColumnName", SqlDbType.VarChar));

            int totalAffected = 0;
            // goes through all coulmns
            foreach (string columnName in columnNames)
            {
                sqlCmd.Parameters["@ColumnName"].Value = columnName;
                affected = sqlCmd.ExecuteNonQuery();
                Debug.WriteLine("Added new column " + columnName + " to table " + tableIdentifier + ":" + affected);
                totalAffected += affected;
            }

            Debug.WriteLine(totalAffected + " columns out of total " + columnNames.Length +
                " have been added to table " + tableIdentifier);
        }


        /// <summary>
        /// Adds table and its columns to the store database.
        /// </summary>
        /// <param name="tableName">The name of tracked source table.</param>
        /// <param name="columnNames">The names of columns of thecked source table.</param>
        public void AddTableAndColumnsToDictionary(string tableName, string[] columnNames)
        {
            // all inputs validation
            if (string.IsNullOrEmpty(tableName))
            {
                throw new ArgumentException("The tableName argument is null or empty.");
            }

            if (columnNames == null || columnNames.Length == 0)
            {
                throw new ArgumentException("The columnNames argument is null or empty.");
            }

            using (SqlConnection sqlConn = new SqlConnection(_storeDbConnStr.ConnectionString))
            {
                // opens connection and begins new transaction
                sqlConn.Open();
                SqlTransaction sqlTran = sqlConn.BeginTransaction();
                try
                {
                    SqlCommand sqlCmd = new SqlCommand();
                    sqlCmd.Connection = sqlConn;
                    sqlCmd.Transaction = sqlTran;

                    AddTableAndColumnsToDictionary(tableName, columnNames, sqlCmd);
                    sqlTran.Commit();
                }
                catch (Exception ex)
                {
                    try
                    {
                        sqlTran.Rollback();
                    }
                    catch
                    {
                        throw new CdcSchemaInstallerException(
                            "The error occured while adding table " + tableName + " and its columns to database, and" +
                            " the transaction could not have been rolled back.", ex);
                    }

                    throw new CdcSchemaInstallerException(
                        "The error occured while adding table " + tableName + " and its columns to database.", ex);
                }

            }
        }
  
        /// <summary>
        /// Installes all necessary sql objects to perform the change data capture process.
        /// </summary>
        /// <param name="tableName">The name of a table to be monitored.</param>
        public void InstallTableTracking(SourceTable sourceTable)
        {
            // current processing step 
            int progress = 0;
            
            // total number of steps in processing
            int maxProgress = 9;

            // validates input arguments
            if (sourceTable == null)
            {
                throw new ArgumentException("The sourceTable argument can not be null.");
            }

            if (sourceTable.PrimaryKey == null)
            {
                throw new CdcSchemaInstallerException("The sourceTable table argument, does not contain" +
                    " primary key.");
            }

            if (!sourceTable.PrimaryKey.ActivateTracking)
            {
                throw new CdcSchemaInstallerException("The Primary Key must be included" + 
                    " into list of tracked columns.");
            }            

            CheckCancel();
            UpdateProgress(progress, "Building scripts.");

            // gets columns and its types for the table being processed
            string[] columnNames = sourceTable.GetToActivateColNames();
            string[] columnTypes = sourceTable.GetToActivateColTypes();

            // builds sql scripts for all elements of cdc system
            string InsertChangedDataSpSql = BuildInsertChangedDataSpSql(sourceTable.TableIdentifier, 
                SqlTemplates.TemplateInsertChangeDataSp);

            string ChangedDataTableSql = BuildChangedDataTableSql(sourceTable.TableIdentifier,
                SqlTemplates.TemplateChangedDataTable);

            string IdentifyChangesSpSql = BuildIdentifyChangesSpSql(sourceTable.TableIdentifier, columnNames,
                SqlTemplates.TemplateIdentifyChangeSp, sourceTable.PrimaryKey.ColumnName);

            string InsertedStageTableSql = BuildInsertedOrDeletedStageTableSql(sourceTable.TableIdentifier, columnNames,
                columnTypes, SqlTemplates.TemplateInsertedStageTable);

            string DeletedStageTableSql = BuildInsertedOrDeletedStageTableSql(sourceTable.TableIdentifier, columnNames,
                columnTypes, SqlTemplates.TemplateDeletedStageTable);

            string CaptureTriggerSql = BuildCaptureTriggerSql(sourceTable.TableIdentifier, sourceTable.TableSchema, 
                sourceTable.TableName, columnNames, sourceTable.GetToActivateBitmask(), SqlTemplates.TemplateCaptureTrigger);

            CheckCancel();
            UpdateProgress(GetPercentProgress(++progress, maxProgress), "Executing scripts.");

            // creates sql command object which will be used to create CDC system schema
            SqlCommand sqlCmd = new SqlCommand();
            sqlCmd.CommandType = CommandType.Text;

#if !DEBUG
            // creates distributed transaction and connection objects for sql scripts execution
            using (TransactionScope distTran = new TransactionScope(TransactionScopeOption.RequiresNew))
#endif
            {
                using (SqlConnection sqlConnMonitor = new SqlConnection(_monitorDbConnStr.ConnectionString),
                                    sqlConnStage = new SqlConnection(_stageDbConnStr.ConnectionString),
                                    sqlConnStore = new SqlConnection(_storeDbConnStr.ConnectionString))
                {
                    // Adds TableAndColumnsToStoreDb
                    sqlConnStore.Open();
                    sqlCmd.Connection = sqlConnStore;
                    AddTableAndColumnsToDictionary(sourceTable.TableIdentifier, columnNames, sqlCmd);
                    sqlCmd.CommandType = CommandType.Text;
                    sqlCmd.Parameters.Clear();

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating change data table.");

                    // Creates ChangedDataStore
                    sqlCmd.CommandText = ChangedDataTableSql;
                    sqlCmd.ExecuteNonQuery();

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating InsertChangedData SP.");

                    // Creates InsertChangedDataSp
                    sqlCmd.CommandText = InsertChangedDataSpSql;
                    sqlCmd.ExecuteNonQuery();

                    // Creates InsertedStageTable
                    sqlConnStage.Open();
                    sqlCmd.Connection = sqlConnStage;

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating InsertedStageTable.");

                    sqlCmd.CommandText = InsertedStageTableSql;
                    sqlCmd.ExecuteNonQuery();

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating DeletedStageTable.");

                    // Creates DeletedStageTable
                    sqlCmd.CommandText = DeletedStageTableSql;
                    sqlCmd.ExecuteNonQuery();

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating IdentifyChanges SP.");

                    // Creates IdentifyChangesSp
                    sqlCmd.CommandText = IdentifyChangesSpSql;
                    sqlCmd.ExecuteNonQuery();

                    CheckCancel();
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Creating CaptureTrigger.");

                    // Creates CaptureTrigger
                    sqlConnMonitor.Open();
                    sqlCmd.Connection = sqlConnMonitor;
                    sqlCmd.CommandText = CaptureTriggerSql;
                    sqlCmd.ExecuteNonQuery();
                    CheckCancel();

                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Committing transaction.");
#if !DEBUG
                    distTran.Complete();
#endif
                    UpdateProgress(GetPercentProgress(++progress, maxProgress), "Completed.");
                }
            }
        }

        /// <summary>
        /// Uninstalls trcking of the table, by dropping the trigger from source table.
        /// </summary>
        /// <param name="tableNames">The table tableIdentifier.</param>
        public void UninstallTableTracking(string tableSchema, string tableIdentifier)
        {
            if (string.IsNullOrEmpty(tableIdentifier))
            {
                throw new ArgumentException("The tableIdentifier argument is null or empty.");
            }

            // prepares sql connection to monitored database and executes trigger dropping command
            using (SqlConnection sqlConn = new SqlConnection(_monitorDbConnStr.ConnectionString))
            {
                SqlCommand sqlCmd = new SqlCommand(
                    BuildDropTriggerSql(tableSchema, tableIdentifier, SqlTemplates.TemplateDropTrigger), sqlConn);
                sqlConn.Open();
                sqlCmd.ExecuteNonQuery();
            }

        }

        /// <summary>
        /// Checks if background worker reqested cancellation, if yes, throw exception to stop processing.
        /// </summary>
        private void CheckCancel()
        {
            if (_backWorker != null && _backWorker.CancellationPending)
            {
                throw new CdcSchemaInstallerCancelException("Installation cancelled.");
            }
        }

        /// <summary>
        /// Updates the progress of a background worker.
        /// </summary>
        /// <param name="percentProgress">Progress value.</param>
        /// <param name="userState">User state, operation currently performed.</param>
        private void UpdateProgress(int percentProgress, object userState)
        {
            // validates input arguments
            if (percentProgress < 0 || percentProgress > 100)
            {
                throw new ArgumentOutOfRangeException("The percentProgress argument out of range.");
            }

            // checks if processing performed using background worker
            if (_backWorker != null)
            {
                if (userState != null)
                {
                    _backWorker.ReportProgress(percentProgress, new AsyncWorkProgressState(
                            AsyncWorkProgressType.ProgressDescription, 
                            new object[]{userState}));
                }
                else
                {
                    _backWorker.ReportProgress(percentProgress);
                }
            }
        }

        /// <summary>
        /// Calculates progress of the operation.
        /// </summary>
        /// <param name="current">Current step of processing. Between 0 and total.</param>
        /// <param name="total">Total number of processing steps. Between 1 and 100.</param>
        /// <returns>Integer progress of operation processing in percents.</returns>
        private int GetPercentProgress(int current, int total)
        {
            // validates input arguments
            if (total < 1 || total > 100)
            {
                throw new ArgumentOutOfRangeException("The total arument out of range.");
            }

            // validates input arguments
            if (current < 0 || current > total)
            {
                throw new ArgumentOutOfRangeException("The current argument out of range.");
            }

            return current * (100 / total);
        }
    }
}
